package com.aotain.metis.bolt;

import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Calendar;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import java.util.Map.Entry;

import org.apache.commons.codec.binary.Base64;
import org.apache.log4j.Logger;
import org.apache.storm.Config;
import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.FailedException;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;

import com.aotain.hbase.HBaseRecordAdd;
import com.aotain.storm.AbstractBolt;

/**
 * @ClassName: CCDetectBolt 
 * @Description:  CC攻击检测方法，HBASE入库
 * @author Turk
 * @date 2016年 
 * 
 * 2016/12/05 目的IP V1 版本copy
 */
public class CCDetectBoltMetis extends AbstractBolt{

	/**
	 * 
	 */
	private static final long serialVersionUID = 1L;

	/**
	 * 每分钟累积MAP  目标|源|URL
	 */
	private HashMap<String,Long> mapUrl = new HashMap<String,Long>();
	
	/**
	 * 上行包字节数
	 */
	private HashMap<String,Long> mapStreamoctets = new HashMap<String,Long>();
	
	/**
	 * 上行包数
	 */
	private HashMap<String,Long> mapStreampacket = new HashMap<String,Long>();

	
	/**
	 * 存放记录连续状态 最后的更新时间
	 */
	private HashMap<String,Long> mapLastTime = new HashMap<String,Long>();
	
	/**
	 * 记录攻击开始时间
	 */
	private HashMap<String,Long> mapAttackStart = new HashMap<String,Long>();
	
	//-----累积包，次数
	/**
	 * 持久-上行包字节数
	 */
	private HashMap<String,Long> mapStreamoctetsMem = new HashMap<String,Long>();
	
	/**
	 * 持久-上行包数
	 */
	private HashMap<String,Long> mapStreampacketMem = new HashMap<String,Long>();

	/**
	 * 持久-攻击次数
	 */
	private HashMap<String,Long> mapCCNumMem = new HashMap<String,Long>();
	
	/**
	 * 持续攻击时记录第一次开始的Map拼接的EntryKey
	 */
	private HashMap<String,String> mapEntryKey = new HashMap<String,String>();
	
	/**
	 * 攻击峰值
	 */
	private HashMap<String,Double> mapHighFlow = new HashMap<String,Double>();
	
	/**
	 * 结束等待时长
	 */
	private final long END_DURATION = 180*1000l;
	
	/**
	 * 连接次数门限 50次/秒
	 */
	private final int ATTACK_THRESHOLD = 50*60;

	
	@Override
	public void execute(Tuple tuple) {
		// TODO Auto-generated method stub
		try {
			if(isTickTuple(tuple)) {
				emitCountingData(collector);
				mapUrl.clear();
				collector.ack(tuple);
			} else {
				countInLocal(tuple); 
				collector.ack(tuple);
			}
		} catch (Exception e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
			collector.fail(tuple);
			throw new FailedException("CCDetectBolt throws an exception!",e);
		}
	}
	
	
	//计算需要的值
	private void countInLocal(Tuple tuple) {
		try
		{
			/*//"dip","dport","sip","sport","gisstr","domainname","url"*/
			String url = tuple.getStringByField("url");	
			if(url==null || url.trim().isEmpty())
				return;
			
			String dip = tuple.getStringByField("dip");
			String sip = tuple.getStringByField("sip");
			String idc = tuple.getStringByField("idcno");
					
			String gis = tuple.getStringByField("gisstr");
			
			String useragent = "useragent";
			String cookie = "cookie";
			if(tuple.contains("useragent") && tuple.contains("cookie"))
			{
				useragent = tuple.getStringByField("useragent");
				cookie = tuple.getStringByField("cookie");
			}
			
			//String gis = new String(Base64.decodeBase64(tuple.getStringByField("gisstr")),"UTF8");			
			String tripleKey = dip + "|" + sip +"|" + url + "|" 
					+ gis + "|" + idc + "|" + useragent + "|" + cookie;
			
			//发现相同URL，相同源IP，+1
			long count = 1L;
			if(mapUrl.containsKey(tripleKey))
			{
				count = mapUrl.get(tripleKey) + 1;
			}
			mapUrl.put(tripleKey, count);
			
			
			//UDII数据特有
			if(tuple.contains("upstreamoctets") && tuple.contains("upstreampacket"))
			{
				//String base64Key = new String(Base64.encodeBase64(tripleKey.getBytes()));
				long nUpStreamOctets = tuple.getLongByField("upstreamoctets");
				long nUpstreampacket = tuple.getLongByField("upstreampacket");
				if(mapStreamoctets.containsKey(tripleKey))
				{
					nUpStreamOctets = mapStreamoctets.get(tripleKey) + nUpStreamOctets;
				}
				mapStreamoctets.put(tripleKey, nUpStreamOctets);
				
				if(mapStreampacket.containsKey(tripleKey))
				{
					nUpstreampacket = mapStreampacket.get(tripleKey) + nUpstreampacket;
				}
				mapStreampacket.put(tripleKey, nUpstreampacket);
			
			}
		}
		catch(Exception ex)
		{
			Logger.getRootLogger().error("=====countInLocal=========ERROR=================",ex);
		}
	}
	
	private void emitCountingData(OutputCollector collector) {
	
		try{
		
		for(Entry<String, Long> entry : mapUrl.entrySet()) {
			//KEY
			String entryKey = entry.getKey();
			//mapWUrl.put(minKey, entry.getValue());
			String[] tuplesp = entryKey.split("\\|",-1);
			String dip = tuplesp[0];
			String sip = tuplesp[1];
			String url = tuplesp[2];
			
			
			String ua = tuplesp[5];
			String cookie = tuplesp[6];
			String hashKey = dip + "_" + sip + "_" + url + "_" + ua + "_" + cookie;
			String base64Key = hashKey;
			//String base64Key = new String(Base64.encodeBase64(hashKey.getBytes()));
			
			//url = new String(Base64.decodeBase64(url));	
			
			if(entry.getValue()>= ATTACK_THRESHOLD)
			{	
				//同一个源IP 到目的IP 相同的URL 在10秒内访问次数超过100次 
				//判断当前源->目的IP是否存在记录
				if(mapLastTime.containsKey(base64Key))
				{
					//存在记录，更新数据
					mapLastTime.put(base64Key, System.currentTimeMillis());
					Logger.getLogger(CCDetectBoltMetis.class).info("###CC V2 ATTACK UPDATE LASTTIME" + hashKey);
					UpdateAttackLog(entry);
				}
				else
				{
					//新发生的攻击行为，记录开始时间并且上报异常攻击日志，但是不上报统计数据
					
					mapLastTime.put(base64Key, System.currentTimeMillis());
					InsertNewAttackLog(entry);
					Logger.getLogger(CCDetectBoltMetis.class).info("###CC V2 ATTACK START" + hashKey);
				}
			}
			else
			{
				if(mapLastTime.containsKey(base64Key))
				{
					//如果存在记录，判断更新时间与当前时间差距是否超过了等待门限
					Long lastTime = mapLastTime.get(base64Key);
					if(System.currentTimeMillis() - lastTime > END_DURATION)
					{
						//超过30秒，认为攻击结束，记录结束时间，
						Logger.getLogger(CCDetectBoltMetis.class).info("###CC V2 ATTACK END" + entryKey);
						EndAttackLog(entryKey);
						//将数据从map中移除
						//mapWUrl.remove(base64Key);
						mapLastTime.remove(base64Key);
						mapStreamoctetsMem.remove(base64Key);
						mapStreampacketMem.remove(base64Key);
						mapCCNumMem.remove(base64Key);
						mapAttackStart.remove(base64Key);
						mapEntryKey.remove(base64Key);
						mapHighFlow.remove(base64Key);
						//Logger.getLogger(CCDetectBoltV2.class).info("###CC V2 ATTACK END" + entryKey);
					}
					else
					{
						//虽然未超过门限，但是在攻击等待范围时间内，将这一分钟的数据累加到攻击数据中。
						UpdateAttackLog(entry);
					}
				}
			}
		}
		
		
		// 对内存中存在的正在监控的攻击数据做核查判断是否定义攻击结束
		ArrayList<String> removeItem = new ArrayList<String>();
		for(Entry<String, Long> entry : mapLastTime.entrySet())
		{
			Long lastTime = entry.getValue();
			String base64Key = entry.getKey();
			Logger.getLogger(CCDetectBoltMetis.class).info(
					String.format("###CC V2 REMOVE hashkey %s time %d",base64Key,(System.currentTimeMillis() - lastTime)/1000));
			
			if(System.currentTimeMillis() - lastTime > END_DURATION)
			{
				//超过等待门限秒，认为攻击结束，记录结束时间，
				String entryKey = mapEntryKey.get(base64Key);
				Logger.getLogger(CCDetectBoltMetis.class).info("###CC V2 ATTACK END" + entryKey);
				EndAttackLog(entryKey);
				//将数据从map中移除
				//mapWUrl.remove(base64Key);
				removeItem.add(base64Key);
				mapStreamoctetsMem.remove(base64Key);
				mapStreampacketMem.remove(base64Key);
				mapCCNumMem.remove(base64Key);
				mapAttackStart.remove(base64Key);
				mapEntryKey.remove(base64Key);
				mapHighFlow.remove(base64Key);
				
			}
		}
		
		//remove 已经攻击结束的记录
		for(String base64Key : removeItem)
		{
			mapLastTime.remove(base64Key);
		}
		
		removeItem.clear();
		mapStreamoctets.clear();
		mapStreampacket.clear();
		mapUrl.clear();
		
		} catch (Exception e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
			throw new FailedException("CCDetectBoltV2 emitCountingData throws an exception!",e);
		}
	}
	
	private void InsertNewAttackLog(Entry<String, Long> entry)
	{
		HBaseRecordAdd hbaseInstance = HBaseRecordAdd.getInstance(zooserver);
		
		String[] tuplesp = entry.getKey().split("\\|",-1);
		String dip = tuplesp[0];
		String sip = tuplesp[1];
		String url = tuplesp[2];
		
		String gis = new String(Base64.decodeBase64(tuplesp[3]));
		String[] gisArray = gis.split("#",-1);
		String idc = tuplesp[4];
		String ua = tuplesp[5];
		String cookie = tuplesp[6];
		
		String destAreaName = gisArray[0];
		String destGis = gisArray[1];
		String sourceAreaName = gisArray[2];
		String sourceGis = gisArray[3];
		String sourceAreaCountry = gisArray[4];
		String sourceAreaId = gisArray[5];							
		String sourceProvinceName =  gisArray[6].isEmpty()?sourceAreaCountry:gisArray[6]; //如果省为空，精确到国家
		String sourceAreaCityId = gisArray[7];
		String sourceAreaProvinceId = gisArray[8];
		
		String hashKey = dip + "_" + sip + "_" + url + "_" + ua + "_" + cookie;
		//String base64Key = new String(Base64.encodeBase64(hashKey.getBytes()));
		String base64Key = hashKey; 
		//url = new String(Base64.decodeBase64(url));	
		
		long starttime = System.currentTimeMillis();
		mapAttackStart.put(base64Key,starttime);//记录开始时间
		
		String accesstime = new SimpleDateFormat("yyyyMMddHHmmss").format(new Date(starttime));
		String rowKeyAbnormal =  accesstime + "_" + sip + "_" + dip +  "_" + "3";	

		String desc = String.format("WEB攻击 ,攻击URL:%s", url);
				
		hbaseInstance.Add("METIS_ABNORMAL_LOG", rowKeyAbnormal, "cf", "SOURCEIP", sip);
		hbaseInstance.Add("METIS_ABNORMAL_LOG", rowKeyAbnormal, "cf", "DESTPORT", "");							
		hbaseInstance.Add("METIS_ABNORMAL_LOG", rowKeyAbnormal, "cf", "ACCESSTIME", accesstime);
		hbaseInstance.Add("METIS_ABNORMAL_LOG", rowKeyAbnormal, "cf", "ABRNORMAL", "3");
		hbaseInstance.Add("METIS_ABNORMAL_LOG", rowKeyAbnormal, "cf", "DESTIP", dip);							
		hbaseInstance.Add("METIS_ABNORMAL_LOG", rowKeyAbnormal, "cf", "SOURCEAREA", sourceAreaName);
		hbaseInstance.Add("METIS_ABNORMAL_LOG", rowKeyAbnormal, "cf", "SOURCEGEO", sourceGis);
		hbaseInstance.Add("METIS_ABNORMAL_LOG", rowKeyAbnormal, "cf", "SOURCECOUNTRY", sourceAreaCountry);
		hbaseInstance.Add("METIS_ABNORMAL_LOG", rowKeyAbnormal, "cf", "DESTAREA", destAreaName);
		hbaseInstance.Add("METIS_ABNORMAL_LOG", rowKeyAbnormal, "cf", "DESTGEO", destGis);							
		hbaseInstance.Add("METIS_ABNORMAL_LOG", rowKeyAbnormal, "cf", "ATTNUM", entry.getValue().toString());
		hbaseInstance.Add("METIS_ABNORMAL_LOG", rowKeyAbnormal, "cf", "DESC", desc);
		hbaseInstance.Add("METIS_ABNORMAL_LOG", rowKeyAbnormal, "cf", "EVALUATE", "40");
		hbaseInstance.Add("METIS_ABNORMAL_LOG", rowKeyAbnormal, "cf", "PROVINCE", sourceProvinceName);

		long countValue = entry.getValue();

		//UDII数据特有
		long nUpStreamOctets = 0;
		long nUpstreampacket = 0;
		
		if(mapStreamoctets.size() > 0 && mapStreampacket.size() > 0)
		{
			nUpStreamOctets = mapStreamoctets.get(entry.getKey());
			nUpstreampacket = mapStreampacket.get(entry.getKey());
			hbaseInstance.Add("METIS_ABNORMAL_LOG", rowKeyAbnormal, "cf", "UPSTREAMOCTETS", String.valueOf(nUpStreamOctets));
			hbaseInstance.Add("METIS_ABNORMAL_LOG", rowKeyAbnormal, "cf", "UPSTREAMPACKET", String.valueOf(nUpstreampacket));
			
			mapStreamoctetsMem.put(base64Key, nUpStreamOctets);
			mapStreampacketMem.put(base64Key, nUpstreampacket);
		}
		mapCCNumMem.put(base64Key, countValue);
		
		mapEntryKey.put(base64Key, entry.getKey());
		
		Logger.getLogger(CCDetectBoltMetis.class).info(
				String.format("###CC V2 ATTACK START ROWKEY:%s DESC:%s  HASHKEY:%s",rowKeyAbnormal, desc, hashKey));
		
		
	}
	
	private void UpdateAttackLog(Entry<String, Long> entry)
	{
		String[] tuplesp = entry.getKey().split("\\|",-1);
		String dip = tuplesp[0];
		String sip = tuplesp[1];
		String url = tuplesp[2];
		
		
		String ua = tuplesp[5];
		String cookie = tuplesp[6];
		String hashKey = dip + "_" + sip + "_" + url + "_" + ua + "_" + cookie;
		String base64Key = hashKey;
		//String base64Key = new String(Base64.encodeBase64(hashKey.getBytes()));
		
		//url = new String(Base64.decodeBase64(url));	
		
		//开始时间
		//String accesstime = new SimpleDateFormat("yyyyMMddHHmmss").format(new Date(mapAttackStart.get(base64Key)));
		//String rowKeyAbnormal = dip + "_" + accesstime + "_" + sip + "_" + "_3";
		
		//UDII数据特有
		long nUpStreamOctets = 0;
		long nUpstreampacket = 0;
		if(mapStreamoctets.size() > 0 && mapStreampacket.size() > 0)
		{
			nUpStreamOctets = mapStreamoctets.get(entry.getKey());
			nUpstreampacket = mapStreampacket.get(entry.getKey());
			if(mapStreamoctetsMem.containsKey(base64Key))
				mapStreamoctetsMem.put(base64Key, mapStreamoctetsMem.get(base64Key) + nUpStreamOctets);
			if(mapStreampacketMem.containsKey(base64Key))
				mapStreampacketMem.put(base64Key, mapStreampacketMem.get(base64Key) + nUpstreampacket);
		}
		
		long countValue = entry.getValue();
		if(mapCCNumMem.containsKey(base64Key))
			mapCCNumMem.put(base64Key, mapCCNumMem.get(base64Key) + countValue);
		
		String desc = String.format("WEB攻击 ,攻击URL[%s]", url);
		double maxValue = (double)nUpStreamOctets/(double)60/(double)1024 ;
		
		
		if(mapHighFlow.containsKey(base64Key))
		{
			double maxValueMem = mapHighFlow.get(base64Key);
			if(maxValueMem < maxValue)
			{
				maxValueMem = maxValue;
				mapHighFlow.put(base64Key, maxValueMem);
			}
			else
			{
				maxValue = maxValueMem;
			}
		}
		else
		{
			mapHighFlow.put(base64Key, maxValue);
		}
		
		desc = String.format("%s 攻击峰值为 :%f KB/s ", desc,maxValue);
		
		HBaseRecordAdd hbaseInstance = HBaseRecordAdd.getInstance(zooserver);
		
		//开始时间
		String accesstime = new SimpleDateFormat("yyyyMMddHHmmss").format(new Date(mapAttackStart.get(base64Key)));
				
		String rowKeyAbnormal =  accesstime + "_" + sip + "_" + dip +  "_" + "3";	
				
		if(mapStreamoctets.size() > 0 && mapStreampacket.size() > 0)
		{
			nUpStreamOctets = mapStreamoctetsMem.get(base64Key);
			nUpstreampacket = mapStreampacketMem.get(base64Key);
			hbaseInstance.Add("METIS_ABNORMAL_LOG", rowKeyAbnormal, "cf", "UPSTREAMOCTETS", String.valueOf(nUpStreamOctets));
			hbaseInstance.Add("METIS_ABNORMAL_LOG", rowKeyAbnormal, "cf", "UPSTREAMPACKET", String.valueOf(nUpstreampacket));
			desc = String.format("%s 攻击流量[%f KB]",desc, (double)nUpStreamOctets/(double)1024);
		}
		
		
		desc = String.format("%s 攻击次数 [%d]",desc, countValue);
		Logger.getLogger(CCDetectBoltMetis.class).info(
				String.format("###CC V2 ATTACK UPDATE ROWKEY:%s DESC:%s  HASHKEY:%s",rowKeyAbnormal, desc, hashKey));
		hbaseInstance.Add("METIS_ABNORMAL_LOG", rowKeyAbnormal, "cf", "DESC", desc);
		hbaseInstance.Add("METIS_ABNORMAL_LOG", rowKeyAbnormal, "cf", "ATTNUM", String.valueOf(countValue));
		
		String gis = new String(Base64.decodeBase64(tuplesp[3]));
		String[] gisArray = gis.split("#",-1);
		String sourceAreaName = gisArray[2];
		String sourceAreaCountry = gisArray[4];
		String sourceAreaId = gisArray[5];							
		String sourceProvinceName =  gisArray[6].isEmpty()?sourceAreaCountry:gisArray[6]; //如果省为空，精确到国家
		String idc = tuplesp[4];
		
		Date current = new Date(System.currentTimeMillis());
		SimpleDateFormat sdf_h  = new SimpleDateFormat("yyyyMMddHH");
		SimpleDateFormat sdf_d  = new SimpleDateFormat("yyyyMMdd");
		SimpleDateFormat sdf_hour  = new SimpleDateFormat("HH:00");
		String dateStr_h = sdf_h.format(current);
		String dateStr_d = sdf_d.format(current);
		String dateStr_hour = sdf_hour.format(current);
		
		nUpStreamOctets = mapStreamoctets.get(entry.getKey());
		nUpstreampacket = mapStreampacket.get(entry.getKey());
		if(nUpStreamOctets != 0)
			countValue = nUpStreamOctets;//UDII数据用于趋势图展示
		
		collector.emit(new Values(dip,sip,"","",dateStr_d, dateStr_h, sourceAreaId, sourceAreaName,
				sourceProvinceName, countValue, dateStr_hour,"CC", idc, sourceAreaCountry,(long)nUpStreamOctets,(long)nUpstreampacket));
		
		
	}
	
	private void EndAttackLog(String entryKey)
	{
		//UpdateAttackLog(entry);


		String[] tuplesp = entryKey.split("\\|",-1);
		String dip = tuplesp[0];
		String sip = tuplesp[1];
		String url = tuplesp[2];
		
		String gis = new String(Base64.decodeBase64(tuplesp[3]));
		String[] gisArray = gis.split("#",-1);
		String idc = tuplesp[4];
		String ua = tuplesp[5];
		String cookie = tuplesp[6];
		
		String sourceAreaName = gisArray[2];
		String sourceAreaCountry = gisArray[4];
		String sourceAreaId = gisArray[5];							
		String sourceProvinceName =  gisArray[6].isEmpty()?sourceAreaCountry:gisArray[6]; //如果省为空，精确到国家
		
		
		String hashKey = dip + "_" + sip + "_" + url + "_" + ua + "_" + cookie;
		String base64Key = hashKey;
//		String base64Key = new String(Base64.encodeBase64(hashKey.getBytes()));
		
		//url = new String(Base64.decodeBase64(url));	
		
		//开始时间
		String accesstime = new SimpleDateFormat("yyyyMMddHHmmss").format(new Date(mapAttackStart.get(base64Key)));
		String endtime = new SimpleDateFormat("yyyyMMddHHmmss").format(new Date(mapLastTime.get(base64Key)+60*1000l));
		
		String rowKeyAbnormal =  accesstime + "_" + sip + "_" + dip +  "_" + "3";	
		HBaseRecordAdd hbaseInstance = HBaseRecordAdd.getInstance(zooserver);
		
		String desc = String.format("WEB攻击 ,攻击URL[%s]", url);
		
		hbaseInstance.Add("METIS_ABNORMAL_LOG", rowKeyAbnormal, "cf", "ATTNUM", String.valueOf(mapCCNumMem.get(base64Key)));
		hbaseInstance.Add("METIS_ABNORMAL_LOG", rowKeyAbnormal, "cf", "ENDTIME", endtime);
		
		desc = String.format("%s 攻击时间[%s->%s] 攻击时长[%s秒]", desc, accesstime, endtime ,
					(mapLastTime.get(base64Key)+60*1000l-mapAttackStart.get(base64Key))/1000);
		
		long nUpStreamOctets = 0;
		long nUpstreampacket = 0;
		if(mapStreamoctets.size() > 0 && mapStreampacket.size() > 0)
		{
			nUpStreamOctets = mapStreamoctetsMem.get(base64Key);
			nUpstreampacket = mapStreampacketMem.get(base64Key);
			hbaseInstance.Add("METIS_ABNORMAL_LOG", rowKeyAbnormal, "cf", "UPSTREAMOCTETS", String.valueOf(nUpStreamOctets));
			hbaseInstance.Add("METIS_ABNORMAL_LOG", rowKeyAbnormal, "cf", "UPSTREAMPACKET", String.valueOf(nUpstreampacket));
			desc = String.format("%s 攻击流量[%f KB]",desc, (double)nUpStreamOctets/(double)1024);
		}
		
		long countValue = mapCCNumMem.get(base64Key);
		desc = String.format("%s 攻击次数 [%d]",desc, countValue);
		
		if(mapHighFlow.containsKey(base64Key))
		{
			double maxValueMem = mapHighFlow.get(base64Key);
			desc = String.format("%s 攻击峰值 [%f KB/s]",desc, maxValueMem);
		}
		hbaseInstance.Add("METIS_ABNORMAL_LOG", rowKeyAbnormal, "cf", "DESC", desc);
		hbaseInstance.Add("METIS_ABNORMAL_LOG", rowKeyAbnormal, "cf", "ATTNUM", String.valueOf(countValue));

		
		Logger.getLogger(CCDetectBoltMetis.class).info(
				String.format("###CC V2 ATTACK END ROWKEY:%s DESC:%s  HASHKEY:%s",rowKeyAbnormal, desc, hashKey));

	}
	

	@Override
	public void cleanup() {
		// TODO Auto-generated method stub
		
	}

	@Override
	public void declareOutputFields(OutputFieldsDeclarer declarer) {
		// TODO Auto-generated method stub
		declarer.declare(new Fields("dip","sip","dport","sport","dateStr_d",
				"dateStr_h","sourceAreaId","sourceAreaName","areaName","pv","dateStr_hour",
				"abntype","idcno","sourcecountry","streamoctets","streampacket"));
		declarer.declareStream("ABNORMAL_LOG", new Fields("rowkey","dip","sip","dport",
				"accesstime","sourcearea","sourcegeo","sourcecountry","destarea","destgeo",
				"evaluate","abnormal","desc","attnum","delaytime","flag"));
		
	}

	@Override
	public Map<String, Object> getComponentConfiguration() {
		// TODO Auto-generated method stub
		Map<String,Object> conf = new HashMap<String,Object>();
		conf.put(Config.TOPOLOGY_TICK_TUPLE_FREQ_SECS,60); //1分钟统计一次
		return conf;
	}


	@Override
	public void Init(Map stormConf, TopologyContext context,
			OutputCollector collector) {
		// TODO Auto-generated method stub
		
	}

}
